Skip to content

add processor chain and add support for model and tool #4093

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Open
wants to merge 5 commits into
base: main
Choose a base branch
from
Open

add processor chain and add support for model and tool #4093

wants to merge 5 commits into from
+3,680 −41

Conversation

ylwu-amzn
Copy link
Collaborator

@ylwu-amzn ylwu-amzn commented Aug 12, 2025 •
edited
Loading

Description

When build tutorial for Agentic RAG with Bedrock OpenAI OSS model, see the converse API returns different result with Bedrock Claude model.

Build post process function with Painless Script is painful. Propose a new processor framework for processing outputs from ML models and tools using chainable processors
Supports multiple built-in processors:

  1. to_string: Converts input to string format
  2. regex_replace: Performs regex-based text replacement
  3. jsonpath_filter: Filters JSON using JsonPath expressions
  4. extract_json: Extracts JSON objects/arrays from text
  5. regex_capture: Captures text using regex groups
  6. remove_jsonpath: Removes elements at specified JsonPath
  7. conditional: Applies different processors based on conditions

Test Example

Create Agent with ListIndexTool.

Configure two output processors for ListIndexTool

  1. Use regex_replace to remove all lines contains index starting with .plugins
  2. Use regex_replace to remove first column row , since the row number will become wrong after removing lines
POST _plugins/_ml/agents/_register
{
  "name": "Test Agent",
  "type": "flow",
  "description": "This is a demo agent",
  "tools": [
    {
      "type": "ListIndexTool",
      "parameters": {
        "output_processors": [
          {
            "type": "regex_replace",
            "pattern": """(?m)^(?!row,health,status).*?,.*?,.*?,\.plugins[^\n]*\n""",
            "replacement": ""
          },
          {
            "type": "regex_replace",
            "pattern": """(?m)^row,|^\d+,""",
            "replacement": ""
          }
        ]
      }
    }
  ]
}

POST _plugins/_ml/agents/VTgbg5kBxVtFlT1nPnn4/_execute
{
  "parameters": {}
}

Sample resplonse

{
  "inference_results": [
    {
      "output": [
        {
          "name": "response",
          "result": """health,status,index,uuid,pri(number of primary shards),rep(number of replica shards),docs.count(number of available documents),docs.deleted(number of deleted documents),store.size(store size of primary and replica shards),pri.store.size(store size of primary shards)
yellow,open,ss4o_logs-otel-2025.09.22,aePQaQvuSmygcLXMZiZMpA,1,1,218380,0,117.8mb,117.8mb
yellow,open,jaeger-service-2025-09-20,W_c332tRQiaYyYA7Q93a5g,1,1,84,7,21.9kb,21.9kb
yellow,open,jaeger-service-2025-09-22,aG2VFQLMSVWcefCBAgo2iQ,1,1,84,0,50kb,50kb
yellow,open,jaeger-service-2025-09-23,rmO1fmMxQI2ovc_n-uHN2w,1,1,84,0,30.4kb,30.4kb
green,open,top_queries-2025.09.22-00440,lMQ-f2xtS4eLbNA19GdD2A,1,0,1755,0,639.2kb,639.2kb
yellow,open,jaeger-service-2025-09-24,zM41TnvyTEWkp1wSCwgPdQ,1,1,84,0,20kb,20kb
green,open,top_queries-2025.09.25-00443,epX66GgQSEeznqQoyglZFg,1,0,218,47,576.2kb,576.2kb
"""
        }
      ]
    }
  ]
}

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

Check List

  • New functionality includes testing.
  • New functionality has been documented.
  • API changes companion pull request created.
  • Commits are signed per the DCO using --signoff.
  • Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

@ylwu-amzn ylwu-amzn had a problem deploying to ml-commons-cicd-env August 12, 2025 00:37 — with GitHub Actions Failure
@ylwu-amzn ylwu-amzn had a problem deploying to ml-commons-cicd-env August 12, 2025 00:37 — with GitHub Actions Failure
@ylwu-amzn ylwu-amzn had a problem deploying to ml-commons-cicd-env August 12, 2025 01:18 — with GitHub Actions Failure
@ylwu-amzn ylwu-amzn had a problem deploying to ml-commons-cicd-env August 12, 2025 01:18 — with GitHub Actions Failure
@ylwu-amzn ylwu-amzn had a problem deploying to ml-commons-cicd-env August 12, 2025 19:42 — with GitHub Actions Failure
@ylwu-amzn ylwu-amzn had a problem deploying to ml-commons-cicd-env August 12, 2025 19:42 — with GitHub Actions Failure
@ylwu-amzn ylwu-amzn temporarily deployed to ml-commons-cicd-env September 19, 2025 09:44 — with GitHub Actions Inactive
@ylwu-amzn ylwu-amzn temporarily deployed to ml-commons-cicd-env September 26, 2025 21:13 — with GitHub Actions Inactive
@ylwu-amzn ylwu-amzn temporarily deployed to ml-commons-cicd-env September 26, 2025 21:13 — with GitHub Actions Inactive
@ylwu-amzn ylwu-amzn temporarily deployed to ml-commons-cicd-env September 26, 2025 21:13 — with GitHub Actions Inactive
@ylwu-amzn ylwu-amzn temporarily deployed to ml-commons-cicd-env September 26, 2025 21:13 — with GitHub Actions Inactive
pyek-bot
pyek-bot reviewed Sep 26, 2025
View reviewed changes
common/src/main/java/org/opensearch/ml/common/utils/ToolUtils.java
Comment on lines +235 to +238
modelTensor = ModelTensor.builder().name(outputKey).dataAsMap((Map) output).build();
} else if (output instanceof List) {
Map<String, Object> resultMap = new HashMap<>();
resultMap.put("output", output);
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we put the output in output field here, but above we just return as is if it is a map, so output key may or may not be there? is this expected

ml-algorithms/src/main/java/org/opensearch/ml/engine/algorithms/agent/MLFlowAgentRunner.java
previousStepListener = nextStepListener;
}
}
// firstTool.run(firstToolExecuteParams, firstStepListener);
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

test code?

ml-algorithms/src/main/java/org/opensearch/ml/engine/algorithms/remote/ConnectorUtils.java
Comment on lines +275 to +281
if (processedOutput instanceof String) {
connector.parseResponse((String) processedOutput, modelTensors, scriptReturnModelTensor);
} else {
connector.parseResponse(processedOutput, modelTensors, scriptReturnModelTensor);
}
Copy link
Collaborator

@pyek-bot pyek-bot Sep 26, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this condition seems wrong, if it is a string, you are casting to string else passing the object as is? then why not just pass it in the first place and check inside the parseResponse method

pyek-bot
pyek-bot reviewed Sep 26, 2025
View reviewed changes
ml-algorithms/src/main/java/org/opensearch/ml/engine/tools/ListIndexTool.java
Comment on lines +162 to +166
if (outputParser != null) {
listener.onResponse((T) outputParser.parse(output));
} else {
listener.onResponse((T) output);
}
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

listener.onResponse((T) (outputParser != null ? outputParser.parse(output) : output));

ml-algorithms/src/main/java/org/opensearch/ml/engine/processor/ProcessorChain.java
return captures.get(0);
}
return captures;
// return String.join(" ", captures); // join results with a space
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

test code?

ml-algorithms/src/main/java/org/opensearch/ml/engine/processor/ProcessorChain.java
}
}
if (captures.size() == 1) {
return captures.get(0);
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: use getFirst()

ml-algorithms/src/main/java/org/opensearch/ml/engine/processor/ProcessorChain.java
Comment on lines +137 to +152
if (condition.startsWith(">") && !condition.startsWith(">=")) {
double threshold = Double.parseDouble(condition.substring(1));
return numValue > threshold;
} else if (condition.startsWith("<") && !condition.startsWith("<=")) {
double threshold = Double.parseDouble(condition.substring(1));
return numValue < threshold;
} else if (condition.startsWith(">=")) {
double threshold = Double.parseDouble(condition.substring(2));
return numValue >= threshold;
} else if (condition.startsWith("<=")) {
double threshold = Double.parseDouble(condition.substring(2));
return numValue <= threshold;
} else if (condition.startsWith("==")) {
double threshold = Double.parseDouble(condition.substring(2));
return Math.abs(numValue - threshold) < 1e-10;
}
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit (readability): we don't need else if as there is a return everywhere

rather:

if { return; }

if { return; }

ml-algorithms/src/main/java/org/opensearch/ml/engine/processor/ProcessorChain.java
Comment on lines +213 to +217
if (replaceAll) {
return p.matcher(text).replaceAll(replacement);
} else {
return p.matcher(text).replaceFirst(replacement);
}
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: can just return, no need else

if () {return;} return;

ml-algorithms/src/main/java/org/opensearch/ml/engine/processor/ProcessorChain.java
Comment on lines +281 to +301
if ("object".equalsIgnoreCase(extractType)) {
if (jsonNode.isObject()) {
return mapper.convertValue(jsonNode, Map.class);
} else {
return defaultValue != null ? defaultValue : input;
}
} else if ("array".equalsIgnoreCase(extractType)) {
if (jsonNode.isArray()) {
return mapper.convertValue(jsonNode, List.class);
} else {
return defaultValue != null ? defaultValue : input;
}
} else { // auto
if (jsonNode.isObject()) {
return mapper.convertValue(jsonNode, Map.class);
} else if (jsonNode.isArray()) {
return mapper.convertValue(jsonNode, List.class);
} else {
return defaultValue != null ? defaultValue : input;
}
}
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same here, let's use early return, really helps with readability, too many if/elseif will cause issues

@pyek-bot
Copy link
Collaborator

Overall looks good, left comments to clean up code. Will test it out more thoroughly.

@pyek-bot pyek-bot mentioned this pull request Sep 30, 2025
Open
5 tasks
jiapingzeng
jiapingzeng approved these changes Sep 30, 2025
View reviewed changes
ml-algorithms/src/main/java/org/opensearch/ml/engine/processor/ProcessorChain.java
try {
return Pattern.matches(regex, strValue);
} catch (Exception e) {
log.warn("Invalid regex in condition: {}", regex);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we throw an error here instead of failing silently?

docs/tutorials/agent_framework/rag/agentic_rag_bedrock_claude.md
}
```

Sampel output
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: typo

docs/tutorials/agent_framework/rag/agentic_rag_bedrock_claude.md
@@ -0,0 +1,543 @@
# 1. Create Model
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we explain what the tutorial is for before jumping into steps?

docs/tutorials/agent_framework/rag/agentic_rag_bedrock_openai_oss.md
@@ -0,0 +1,633 @@
# 1. Create Model
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same feedback, could we explain what this tutorial does?

docs/tutorials/agent_framework/rag/agentic_rag_bedrock_claude.md
{
"name": "Query DSL Translator Agent",
"type": "flow",
"description": "This is a demo agent for translating NLQ to OpenSearcdh DSL",
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: typo

docs/tutorials/agent_framework/rag/agentic_rag_bedrock_openai_oss.md
}
```

Sampel output
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: typo

docs/tutorials/agent_framework/rag/agentic_rag_bedrock_openai_oss.md
{
"name": "Query DSL Translator Agent",
"type": "flow",
"description": "This is a demo agent for translating NLQ to OpenSearcdh DSL",
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: typo

@jiapingzeng
Copy link
Contributor

Overall LGTM, have some smaller comments.

@pyek-bot
Copy link
Collaborator

@ylwu-amzn let's separate the tutorial and the PR? Can you raise a separate PR with the tutorial changes?

@ylwu-amzn ylwu-amzn requested a deployment to ml-commons-cicd-env September 30, 2025 06:21 — with GitHub Actions In progress
@ylwu-amzn ylwu-amzn requested a deployment to ml-commons-cicd-env September 30, 2025 06:21 — with GitHub Actions In progress
@ylwu-amzn ylwu-amzn requested a deployment to ml-commons-cicd-env September 30, 2025 06:21 — with GitHub Actions In progress
@ylwu-amzn ylwu-amzn requested a deployment to ml-commons-cicd-env September 30, 2025 06:21 — with GitHub Actions In progress
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@pyek-bot pyek-bot pyek-bot left review comments

@b4sjoo b4sjoo Awaiting requested review from b4sjoo b4sjoo is a code owner

@dhrubo-os dhrubo-os Awaiting requested review from dhrubo-os dhrubo-os is a code owner

@mingshl mingshl Awaiting requested review from mingshl mingshl is a code owner

@jngz-es jngz-es Awaiting requested review from jngz-es jngz-es is a code owner

@model-collapse model-collapse Awaiting requested review from model-collapse model-collapse is a code owner

@rbhavna rbhavna Awaiting requested review from rbhavna rbhavna is a code owner

@zane-neo zane-neo Awaiting requested review from zane-neo zane-neo is a code owner

@Zhangxunmt Zhangxunmt Awaiting requested review from Zhangxunmt Zhangxunmt is a code owner

@austintlee austintlee Awaiting requested review from austintlee austintlee is a code owner

@HenryL27 HenryL27 Awaiting requested review from HenryL27 HenryL27 is a code owner

@sam-herman sam-herman Awaiting requested review from sam-herman sam-herman is a code owner

@xinyual xinyual Awaiting requested review from xinyual xinyual is a code owner

+1 more reviewer

@jiapingzeng jiapingzeng jiapingzeng approved these changes

Reviewers whose approvals may not affect merge requirements

At least 2 approving reviews are required to merge this pull request.

Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
3 participants
@ylwu-amzn @pyek-bot @jiapingzeng